package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.beans.KeywordStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Author: Felix
 * Date: 2021/12/10
 * Desc: 关键词统计
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //注册函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);
        /*
        //TODO 2.设置检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setStateBackend(new FsStateBackend("xxxx"));
        System.setProperty("HADOOP_USER_NAME","");
        */
        //TODO 3.从Kafka的dwd_page_log中读取数据创建动态表
        String pageViewSourceTopic = "dwd_page_log";
        String groupId = "keyword_stats_group";
        tableEnv.executeSql("CREATE TABLE page_view " +
            "(common MAP<STRING,STRING>, " +
            "page MAP<STRING,STRING>,ts BIGINT, " +
            "rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss')) ," +
            "WATERMARK FOR  rowtime  AS  rowtime - INTERVAL '2' SECOND) " +
            "WITH ("+ MyKafkaUtil.getDDL(pageViewSourceTopic,groupId)+")");


        //TODO 4.对数据进行过滤，过滤出搜索行为
        Table fullwordView = tableEnv.sqlQuery("select page['item'] fullword ," +
            "rowtime from page_view  " +
            "where page['page_id']='good_list' " +
            "and page['item'] IS NOT NULL ");


        //TODO 5.对搜索的内容进行分词，并和原表进行连接
        Table keywordView = tableEnv.sqlQuery("select keyword,rowtime  from " + fullwordView + " ," +
            " LATERAL TABLE(ik_analyze(fullword)) as T(keyword)");


        //TODO 6.分组、开窗、聚合计算
        Table keywordStatsSearch  = tableEnv.sqlQuery("select keyword,count(*) ct, '"
            + GmallConstant.KEYWORD_SEARCH + "' source ," +
            "DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt," +
            "DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," +
            "UNIX_TIMESTAMP()*1000 ts from   "+keywordView
            + " GROUP BY TUMBLE(rowtime, INTERVAL '10' SECOND ),keyword");


        //TODO 7.将动态表转换为流
        DataStream<KeywordStats> keywordStatsDS = tableEnv.toAppendStream(keywordStatsSearch, KeywordStats.class);

        //TODO 8.将流中的数据写到ClickHouse中
        keywordStatsDS.print(">>>");
        keywordStatsDS.addSink(
            ClickHouseUtil.getSinkFunction("insert into keyword_stats_0609(keyword,ct,source,stt,edt,ts) values(?,?,?,?,?,?)")
        );
        env.execute();
    }
}
